package com.szwistar.common.comm.jsonrpc.udp;

import java.net.InetSocketAddress;
import java.net.SocketAddress;

import org.apache.log4j.PropertyConfigurator;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.szwistar.common.comm.AsyncResult;
import com.szwistar.common.comm.CommEndpoint;
import com.szwistar.common.comm.InetSocketAddressEx;
import com.szwistar.common.comm.P2pBase;
import com.szwistar.common.comm.blocktrans.BlockTransFrameCodec;

public class JsonRpcStub implements CommEndpoint {
	static final Logger LOG = LoggerFactory.getLogger(JsonRpcStub.class);

	// 记住发送的指令，用于返回结果的比较
	JsonRpcRequest lastRequest = null;
	// 保存调用的Future，以便收到结果后结束等待
	AsyncResult pendingResult = null;

	static {
		// 注册净荷类型
		BlockTransFrameCodec.registerPayloadType(JsonRpcRequest.TYPEID, JsonRpcRequest.class);
		BlockTransFrameCodec.registerPayloadType(JsonRpcResult.TYPEID, JsonRpcResult.class);
	}

	public static JsonRpcStub connectTcp(InetSocketAddress address, Object context) throws Exception {
		P2pBase<JsonRpcStub> p2p = new P2pBase<JsonRpcStub>(address, JsonRpcStub.class, BlockTransFrameCodec.newFilter(), true);
		return p2p.open(context);
	}

	public static JsonRpcStub connectUdp(InetSocketAddress address, Object context) throws Exception {
		P2pBase<JsonRpcStub> p2p = new P2pBase<JsonRpcStub>(address, JsonRpcStub.class, BlockTransFrameCodec.newFilter(), false);
		return p2p.open(context);
	}

	// Session
	protected IoSession session;
	// Owner
	protected JsonRpcStub owner;
	// Server 提供的访问凭证
	long token;

	@Override
	public IoSession getSession() {
		return session;
	}

	@Override
	public boolean isConnected() {
		if(session == null) { return false; }
		return session.isConnected();
	}

	@Override
	public void close() {
		if(session == null) { return; }
		if(session.isConnected()) {	session.close(true); }
		session = null;
	}

	@Override
	public SocketAddress getLocalAddress() {
		return session.getLocalAddress();
	}

	@Override
	public SocketAddress getPeerAddress() {
		return session.getRemoteAddress();
	}

	/**
	 * 发送RPC请求数据帧
	 * @param message
	 * @return
	 */
	@Override
	public WriteFuture write(Object message) {
		return session.write(message);
	}

	public AsyncResult<JsonRpcResult> call(JsonRpcRequest req) {
		if(lastRequest != null) {
			LOG.warn("上一次RPC调用的尚未返回，发起新的调用会丢失上一次的结果: {}", lastRequest.toString());
		}

		// 记住发送的指令，用于返回结果的比较
		lastRequest = req;

		// 设置 token
		req.setToken(token);

		// 发送指令
		write(req);

		pendingResult = new AsyncResult<JsonRpcResult>(1000);
		return pendingResult;
	}

	/**
	 * 调用服务 API 接口
	 * @param service 服务名称
	 * @param method API 接口方法名
	 * @param params 变长参数
	 * @return
	 */
	public AsyncResult<JsonRpcResult> call(String service, String method, Object ...params) {
		JsonRpcRequest req = new JsonRpcRequest(token, service, method, params);
		return call(req);
	}

	/**
	 * 调用对象 API 接口
	 * @param objId 对象ID
	 * @param method API 接口方法名
	 * @param params 变长参数
	 * @return
	 */
	public AsyncResult<JsonRpcResult> call(int objId, String method, Object ...params) {
		JsonRpcRequest req = new JsonRpcRequest(token, objId, method, params);
		return call(req);
	}

	@Override
	public void onCreate(IoSession session, Object owner) throws Exception {
		this.session = session;
		this.owner = (JsonRpcStub) owner;
	}

	@Override
	public void onOpened() {
	}

	@Override
	public void onClosed() {
	}

	@Override
	public void onMessageSent(Object message) {
		LOG.debug("RPC指令：{}", message.toString());
	}

	@Override
	public void onMessageReceived(Object message) {
		LOG.debug("RPC结果：{}", message.toString());
		JsonRpcResult ret = (JsonRpcResult) message;

		if( (lastRequest != null) && (pendingResult != null) &&
			(lastRequest.getSn() == ret.getSn()) &&
			((lastRequest.getToken() == 0) || (lastRequest.getToken() == ret.getToken())) )
		{
			// 返回与命令匹配上了，通知结束等待
			pendingResult.doneWithResult(0, null, ret);
			pendingResult = null;
			lastRequest = null;
			return;
		}
	}

	@Override
	public void onIdle(IdleStatus status) {
	}

	@Override
	public void onExceptionCaught(Throwable cause) {
		LOG.error("JsonRpcStub 遇到异常：", cause);
	}

	public static boolean TEST = false;

	public static void main(String[] args) throws Exception {
		if(args.length < 4) {
			System.out.println("usage: rpc udp 192.168.0.196:9488 5019 toggle");
			return;
		}

		// 加载日志配置文件（放在jar同级的conf目录下）
		PropertyConfigurator.configure("log4j.properties");

		JsonRpcStub rpc = null;
		if(args[0].equals("tcp")) {
			rpc = JsonRpcStub.connectTcp(InetSocketAddressEx.fromString(args[1]), null);
		} else if(args[0].equals("udp")) {
			rpc = JsonRpcStub.connectUdp(InetSocketAddressEx.fromString(args[1]), null);
		} else {
			System.out.println("Bad protocol: " + args[0]);
		}

		Integer objId = null;
		AsyncResult ret = null;

		try { objId = Integer.parseInt(args[2]); } catch(Exception e) {}
		if(objId != null) {
			// 根据参数个数进行调用
			switch(args.length-4) {
			case  0: ret = rpc.call(objId, args[3]); break;
			case  1: ret = rpc.call(objId, args[3], args[4]); break;
			case  2: ret = rpc.call(objId, args[3], args[4], args[5]); break;
			case  3: ret = rpc.call(objId, args[3], args[4], args[5], args[6]); break;
			case  4: ret = rpc.call(objId, args[3], args[4], args[5], args[6], args[7]); break;
			case  5: ret = rpc.call(objId, args[3], args[4], args[5], args[6], args[7], args[8]); break;
			case  6: ret = rpc.call(objId, args[3], args[4], args[5], args[6], args[7], args[8], args[9]); break;
			case  7: ret = rpc.call(objId, args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10]); break;
			case  8: ret = rpc.call(objId, args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10], args[11]); break;
			case  9: ret = rpc.call(objId, args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10], args[11], args[12]); break;
			default: throw new IllegalArgumentException("RPC 参数数量不能超过9个！");
			}
		} else {
			switch(args.length-4) {
			case  0: ret = rpc.call(args[1], args[3]); break;
			case  1: ret = rpc.call(args[1], args[3], args[4]); break;
			case  2: ret = rpc.call(args[1], args[3], args[4], args[5]); break;
			case  3: ret = rpc.call(args[1], args[3], args[4], args[5], args[6]); break;
			case  4: ret = rpc.call(args[1], args[3], args[4], args[5], args[6], args[7]); break;
			case  5: ret = rpc.call(args[1], args[3], args[4], args[5], args[6], args[7], args[8]); break;
			case  6: ret = rpc.call(args[1], args[3], args[4], args[5], args[6], args[7], args[8], args[9]); break;
			case  7: ret = rpc.call(args[1], args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10]); break;
			case  8: ret = rpc.call(args[1], args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10], args[11]); break;
			case  9: ret = rpc.call(args[1], args[3], args[4], args[5], args[6], args[7], args[8], args[9], args[10], args[11], args[12]); break;
			default: throw new IllegalArgumentException("RPC 参数数量不能超过9个！");
			}
		}

		TEST = true;

		System.out.println("等待RPC返回结果...");
		long now = System.currentTimeMillis();

		ret.get();

		now = System.currentTimeMillis() - now;
		System.out.printf("正常结束(耗时: %d ms).\n", now);

		rpc.close();
		System.exit(0);
	}
}
